#ifndef __M_CONNECTION_H__
#define __M_CONNECTION_H__


#include "mq_channel.hpp"
#include "mq_worker.hpp"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/net/TcpClient.h"
#include "muduo/base/CountDownLatch.h"

#include <functional>

namespace wsmq
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        Connection(const std::string &serverip, int serverport, const AsyncWorker::ptr &worker) : 
            _latch(1), _client(worker->_loopthread.startLoop(),
            muduo::net::InetAddress(serverip, serverport), "Client"),
            _dispatcher(std::bind(&Connection::OnUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
            _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
            _worker(worker),
            _channel_manager(std::make_shared<ChannelManager>())
        {
            // 注册业务请求处理函数
            _dispatcher.registerMessageCallback<wsmq::basicCommonResponse>(std::bind(&Connection::basicResponse,
                                                                                     this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::basicConsumeResponse>(std::bind(&Connection::ConsumeResponse,
                                                                                      this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 设置消息回调处理函数
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage,
                                                 _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 设置连接回调处理函数
            _client.setConnectionCallback(std::bind(&Connection::Onconnection, this, std::placeholders::_1));
        
            _client.connect();
            _latch.wait(); // 阻塞等待直到连接建立成功
        }

        void connect()
        {
            _client.connect();
            _latch.wait();
        }

        Channel::ptr openChannel()
        {
            // if (!_conn || !_conn->connected()) {
            //     LOG_ERROR << "Connection not ready";
            //     return Channel::ptr();
            // }
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            bool ret = channel->openChannel();
            if(ret == false)
            {
                DLOG("打开信道失败");
                return Channel::ptr();
            }
            return channel;
        }
        void closeChannel(const Channel::ptr& channel)
        {
            channel->closeChannel();
            _channel_manager->remove(channel->cid());
        }
    private:
        void OnUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            _conn->shutdown();
        }
        void basicResponse(const muduo::net::TcpConnectionPtr &conn, const basicCommonResponsePtr &message, muduo::Timestamp)
        {
            // 找到信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("未找到信道消息");
                return ;
            }
            // 将响应对象添加到信道对应的哈希表中
            channel->putBasicResponse(message);
        }
        void ConsumeResponse(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr &message, muduo::Timestamp)
        {
            // 找到信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("未找到信道消息");
                return ;
            }
            // 封装异步消息处理任务，抛入线程池
            _worker->_threadpool.Push([channel, message](){
                channel->consume(message);
            });
        }
        void Onconnection(const ::muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _conn = conn;
                _latch.countDown();
            }
            else
            {
                _conn.reset();
            }
        }
    private:
        muduo::CountDownLatch _latch;
        muduo::net::TcpConnectionPtr _conn;
        muduo::net::TcpClient _client;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        AsyncWorker::ptr _worker;
        ChannelManager::ptr _channel_manager;
    };
}

#endif